// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/lmax_digital/feed_v1/processor.h"

#include <string>
#include <string_view>
#include <set>
#include <utility>

#include "quickfix/fix42/MarketDataSnapshotFullRefresh.h"
#include "quickfix/fix42/MarketDataIncrementalRefresh.h"

#include "coin2/exchange/lmax_digital/feed_v1/fix_connection.h"

namespace coin2::exchange::lmax_digital::feed_v1::impl {

int64_t DateTimeToNanosecond(const FIX::DateTime& dt) {
  int64_t nano = 0;
  if (dt.m_date > 0) {
    nano += dt.NANOS_PER_SEC * dt.SECONDS_PER_DAY * (dt.m_date - dt.JULIAN_19700101);
  }

  if (dt.m_time > 0) {
    nano += dt.m_time;
  }

  return nano;
}

void LmaxdigitalFeedProcessor::PushEntry(
    const FIX::MDUpdateAction& action,
    const FIX::MDEntryPositionNo& no,
    const PriceQty& price_qty,
    std::vector<PriceQty>* entries) {
  if (action == FIX::MDUpdateAction_NEW) {
    entries->insert(entries->begin() + (no.getValue() - 1), price_qty);
  } else if (action == FIX::MDUpdateAction_CHANGE){
    entries->insert(entries->begin() + (no.getValue() - 1), price_qty);
  } else if (action == FIX::MDUpdateAction_DELETE){
    entries->erase(entries->begin() + (no.getValue() - 1));
  } else {
    CHECK(false) << "Unkown MDEntryType " << action;
  }

  entries->resize(market_depth_, {0, 0});
}

void LmaxdigitalFeedProcessor::UpdateFeed(
    const std::string& native_symbol,
    int64_t feed_time,
    bool book_update) {
  if (book_update) {
    BookEntry& book_entry = book_cache_[native_symbol];
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    {
      auto tx = book->NewTransaction(feed_time);
      tx.Clear();
      for (auto& bid : book_entry.bids) {
        tx.UpdateBid(bid.price, bid.qty);
      }

      for (auto& ask : book_entry.asks) {
        tx.UpdateAsk(ask.price, ask.qty);
      }
    }

    upd_mgr_.PublishBook();
  }

  if (!trades_.empty()) {
    int trade_idx = 0;
    int last_idx = trades_.size() - 1;
    for (auto& trade_entry : trades_) {
      auto* trade = upd_mgr_.GetTradeMutable();
      trade->timestamp = trade_entry.time;
      trade->price = trade_entry.trade.price;
      trade->fill_qty = trade_entry.trade.qty;
      trade->side = coin2::exchange::feed::TRADE_NO_SIDE;
      upd_mgr_.PublishTrade(native_symbol, trade_idx == last_idx);
      trade_idx++;
    }
  }
}

LmaxdigitalFeedProcessor::BookEntry& LmaxdigitalFeedProcessor::BookCache(
    const std::string& native_symbol) {
  BookEntry& book = book_cache_[native_symbol];
  book.asks.resize(market_depth_, {0, 0});
  book.bids.resize(market_depth_, {0, 0});
  return book;
}

LmaxdigitalFeedProcessor::LmaxdigitalFeedProcessor(
      const std::string& system_name,
      const coin2::exchange::di::ConstantInjection& ci,
      const feed::RawParserConfig& config,
      const MarketExchangeApi& mea)
      : FeedParsingProcessor(
            system_name,
            mea,
            ci,
            config) {
  const std::string& host = ci.GetFeedSubscriptionSpec(mea).host;
  constexpr std::string_view fix_protocol("fix://");
  CHECK(host.rfind(fix_protocol) == 0)
      << "Not FIX config " << host;

  std::string config_path = host.substr(fix_protocol.size());
  FIX::SessionSettings settings(config_path);
  std::set<FIX::SessionID> session_ids = settings.getSessions();
  for (auto& session_id : session_ids) {
    market_depth_ = settings.get(session_id).getInt("MarketDepth");
    break;
  }

  dict_ = FIX::DataDictionary("data/coin2/feed/ops/feed-writer/fix/FIX42.xml");
}

void LmaxdigitalFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  FIX::Message message(packet.payload, dict_);
  FIX::MsgType msg_type;
  if (!message.getHeader().getFieldIfSet(msg_type)) {
    if (parser_config_.continue_on_error()) {
      LOG(ERROR) << "[LmaxdigitalFeedProcessor] unknown raw msg: " << packet.payload;
    } else {
      throw std::runtime_error("Invalid FIX Message :" + packet.payload);
    }
    return;
  }

  bool book_enable = CheckChannelFilter("book");
  bool trade_enable = CheckChannelFilter("trade");

  FIX::SendingTime sending_time;
  message.getHeader().getFieldIfSet(sending_time);
  int64_t nano = DateTimeToNanosecond(sending_time);
  bool book_update = false;
  trades_.clear();
  if (msg_type == FIX::MsgType_MarketDataSnapshotFullRefresh) {
    FIX::Symbol symbol;
    std::string native_symbol = message.getField(symbol.getTag());
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }

    FIX42::MarketDataSnapshotFullRefresh::NoMDEntries entry_group;
    FIX::NoMDEntries num;
    FIX::MDEntryType entry_type;
    FIX::MDEntryPx entry_px;
    FIX::MDEntrySize entry_size;
    FIX::MDEntryPositionNo entry_pos;
    message.getField(num);
    if (num == 0) return;  // No data

  BookEntry& book = BookCache(native_symbol);
    for (int idx = 1; idx <= num; idx++) {
      message.getGroup(idx, entry_group);
      entry_group.get(entry_type);
      entry_group.get(entry_px);
      entry_group.get(entry_size);
      entry_group.getFieldIfSet(entry_pos);
      PriceQty price_qty(entry_px.getValue(), entry_size.getValue());
      if (entry_type == FIX::MDEntryType_BID) {
        if (!book_enable) continue;

        PushEntry(FIX::MDUpdateAction_NEW, entry_pos, price_qty, &(book.bids));
        book_update = true;
      } else if (entry_type == FIX::MDEntryType_OFFER) {
        if (!book_enable) continue;

        PushEntry(FIX::MDUpdateAction_NEW, entry_pos, price_qty, &(book.asks));
        book_update = true;
      } else if (entry_type == FIX::MDEntryType_TRADE) {
        if (!trade_enable) continue;

        trades_.emplace_back(TradeEntry{price_qty, nano});
      }
    }

    UpdateFeed(native_symbol, nano, book_update);
  } else if (msg_type == FIX::MsgType_MarketDataIncrementalRefresh) {
    FIX42::MarketDataIncrementalRefresh::NoMDEntries entry_group;
    FIX::NoMDEntries num;
    FIX::MDUpdateAction action;
    FIX::MDEntryType entry_type;
    FIX::Symbol symbol;
    FIX::MDEntryPositionNo entry_pos;
    message.getField(num);
    if (num == 0) return;  // No data

    // suppose 1 message only contains 1 symbol
    message.getGroup(1, entry_group);
    entry_group.get(symbol);
    std::string native_symbol = symbol;
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }
    BookEntry& book = BookCache(native_symbol);
    for (int idx = 1; idx <= num; idx++) {
      message.getGroup(idx, entry_group);

      entry_group.get(symbol);
      CHECK_EQ(native_symbol, symbol);

      entry_group.get(action);
      entry_group.get(entry_type);
      FIX::MDEntryPx entry_px;
      FIX::MDEntrySize entry_size;
      PriceQty price_qty(0, 0);
      if (entry_group.getFieldIfSet(entry_px) && entry_group.getFieldIfSet(entry_size)) {
        price_qty = PriceQty(entry_px.getValue(), entry_size.getValue());
      }

      if (entry_type == FIX::MDEntryType_BID) {
        if (!book_enable) continue;

        entry_group.get(entry_pos);
        PushEntry(action, entry_pos, price_qty, &(book.bids));
        book_update = true;
      } else if (entry_type == FIX::MDEntryType_OFFER) {
        if (!book_enable) continue;

        entry_group.get(entry_pos);
        PushEntry(action, entry_pos, price_qty, &(book.asks));
        book_update = true;
      } else if (entry_type == FIX::MDEntryType_TRADE) {
        if (!trade_enable) continue;

        FIX::MDEntryDate entry_date;
        FIX::MDEntryTime entry_time;
        entry_group.get(entry_date);
        entry_group.get(entry_time);
        int64_t trade_time = DateTimeToNanosecond(entry_date) +
                             DateTimeToNanosecond(entry_time);
        trades_.emplace_back(TradeEntry{price_qty, trade_time});
      }
    }
    UpdateFeed(native_symbol, nano, book_update);
  }

  upd_mgr_.onEndPacket(packet);
}  // class LmaxdigitalFeedProcessor

}  // namespace coin2::exchange::lmax_digital::feed_v1::impl
